package com.aiwiown.snackmq.broker.handler;

import com.aiwiown.snackmq.broker.auth.AuthorizationService;
import com.aiwiown.snackmq.common.auth.Action;
import com.aiwiown.snackmq.common.exception.AuthorizationException;
import com.aiwiown.snackmq.common.handler.MessageHandler;
import com.aiwiown.snackmq.common.handler.RequestContext;
import com.aiwiown.snackmq.common.message.Message;
import com.aiwiown.snackmq.common.message.MessageStatus;
import com.aiwiown.snackmq.common.message.MessageType;
import com.aiwiown.snackmq.storage.transaction.TransactionLog;
import com.aiwiown.snackmq.storage.transaction.TransactionRecord;
import com.aiwiown.snackmq.storage.transaction.TransactionState;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;

/**
 * 处理事务消息的第二阶段：回滚事务。
 */
@Slf4j
@RequiredArgsConstructor
public class RollbackMessageHandler implements MessageHandler {

    private final TransactionLog transactionLog;
    private final AuthorizationService authorizationService;

    @Override
    public MessageType getMessageType() {
        return MessageType.ROLLBACK_TRANSACTIONAL_MESSAGE;
    }

    @Override
    public void handle(RequestContext ctx, Message request) {
        String transactionId = request.getProperty("transactionId");
        if (transactionId == null || transactionId.isEmpty()) {
            ctx.writeResponse(createErrorResponse(request, "TransactionId is missing."));
            return;
        }

        // 即使是回滚，也需要获取半消息来检查权限
        Message halfMessage = transactionLog.getPendingMessage(transactionId);
        if (halfMessage == null) {
            log.warn("Cannot rollback transaction '{}': it may have already been resolved or is invalid.", transactionId);
            // 即使找不到事务，也返回成功，因为最终状态是一致的（事务不存在）
            ctx.writeResponse(createSuccessResponse(request));
            return;
        }

        try {
            // 授权检查：回滚事务也需要对原始主题有生产权限
            // 【修复】使用新的 Resource 对象进行授权，明确资源类型为 TOPIC
            authorizationService.authorize(request.getProperty("authToken"), Action.PRODUCE, new AuthorizationService.Resource("TOPIC", halfMessage.getTopic()));
            // 【修复】从半消息中获取 producerId
            String producerId = halfMessage.getProperty("producerId");
            // 将回滚记录写入事务日志
            TransactionRecord rollbackRecord = new TransactionRecord(
                    transactionId,
                    TransactionState.ROLLED_BACK,
                    System.currentTimeMillis(),
                    null, // 回滚记录不需要消息体
                    producerId // 【修复】传递 producerId
            );
            transactionLog.append(rollbackRecord);

            ctx.writeResponse(createSuccessResponse(request));
            log.info("Transaction {} rolled back successfully.", transactionId);

        } catch (AuthorizationException e) {
            log.warn("Authorization failed for rollback transaction request: {}", e.getMessage());
            ctx.writeResponse(createErrorResponse(request, "Permission denied: " + e.getMessage()));
        } catch (IOException e) {
            log.error("Failed to write ROLLBACK record for transaction {}. The transaction may remain in a pending state.", transactionId, e);
            ctx.writeResponse(createErrorResponse(request, "Failed to finalize transaction state. Please check broker logs."));
        }
    }

    private Message createSuccessResponse(Message request) {
        return Message.builder().messageId(request.getMessageId()).type(MessageType.RESPONSE).status(MessageStatus.SUCCESS).build();
    }

    private Message createErrorResponse(Message request, String errorMessage) {
        return Message.builder().messageId(request.getMessageId()).type(MessageType.RESPONSE).status(MessageStatus.FAILED).errorMessage(errorMessage).build();
    }
}